package com.pw.study.flink.sink;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;

public class MyJdbcSink {

    public static void main(String[] args) {
        dealMysql();
    }

    private static void dealMysql() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("2", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("3", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("4", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("5", 1607527995000L, 30));
        env.fromCollection(waterSensors).addSink(JdbcSink.sink(
                "replace into sensor(id, ts, vc)values(?,?,?)", new JdbcStatementBuilder<WaterSensor>() {
                    @Override
                    public void accept(PreparedStatement ps, WaterSensor sensor) throws SQLException {
                        ps.setString(1, sensor.getId());
                        ps.setLong(2, sensor.getTs());
                        ps.setInt(3, sensor.getVc());
                    }
                }, new JdbcExecutionOptions.Builder().withBatchIntervalMs(3000)
                        .withBatchSize(16 * 1024).withMaxRetries(3).build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUrl("jdbc:mysql://hadoop162:3306/test?useSSL=false")
                        .withUsername("root")
                        .withPassword("aaaaaa")
                        .build()
        ));


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
